package ws.very.util.akka.remote

import akka.actor.OneForOneStrategy
import akka.actor.Props
import akka.actor.Terminated
import akka.actor.ActorLogging
import java.util.UUID
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.Deploy
import akka.actor.SupervisorStrategy
import akka.remote.RemoteScope
import scala.concurrent.duration._


case class Ack(id: String)
class Receiver(target: ActorRef) extends Actor with ActorLogging {
  context.watch(target)

  def receive = {
    case proxy @ Msg(id, msg, snd) =>
      sendAndAck(proxy)
    case MsgWithTimeout(msg, sendTime, timeout) =>
      val now = System.currentTimeMillis
      if ((now - sendTime) * 2 < timeout)
        sendAndAck(msg)
      else println("计算时间不满足 消息被取消")

    case t @ Terminated(`target`) =>
      //sender ! t
      context stop self
  }

  def sendAndAck(proxymsg: Msg) = {
    val Msg(id, msg, snd) = proxymsg
    target.tell(msg, snd)
    sender ! Ack(id)
  }

}

case class Msg(id: String, msg: Any, sender: ActorRef)

/**如果集群时间不准，将有问题**/
case class MsgWithTimeout(msg: Msg, sendTime: Long, timeout: Long)
case object CheckSend
case object Reconn
object AckProxyAc {
  def receiver(target: ActorRef): Props = Props(new Receiver(target))
}

abstract class AckProxyAc(reconnInterval: Long, checkMsgInterval: Long, msgTimeout: Long, target: ActorRef) extends Actor with ActorLogging {

  override def supervisorStrategy = OneForOneStrategy() {
    case _ => SupervisorStrategy.Escalate
  }
  case class SendCache(msg: Msg, initTime: Long, lastSendTime: Option[Long], lastTarget: Option[String])
  var cache = List[SendCache]()

  var tunnel: ActorRef = null
  var maybeOnline = true
  def createTunnel = {
    tunnel = context.actorOf(AckProxyAc.receiver(target).withDeploy(Deploy(scope = RemoteScope(target.path.address))), target.path.elements.mkString("_") + "_ackProxy")
    context.watch(tunnel)
    maybeOnline = true
  }
  def receive = {
    case Terminated(t) if t == tunnel =>
      maybeOnline = false
      startReConnTimer
    case Reconn => createTunnel
    case CheckSend =>
      handleCheckSend
      startCheckSendTimer
    case Ack(id: String) if sender == tunnel =>
      cache = cache filterNot { case SendCache(Msg(i, _, _), _, _, _) => i == id }
    case msg: Any =>
      val time = System.currentTimeMillis()
      val id = UUID.randomUUID().toString()
      val proxyMsg = Msg(id, msg, sender)
      if (!maybeOnline /*|| !cache.isEmpty*/ ) {
        cache = cache ::: List(SendCache(proxyMsg, time, None, None))
        // cache = cache ::: List((id -> (proxyMsg, (time, None))))
      } else {
        sendProxyMsg(proxyMsg, time)
        //cache = cache ::: List((id -> (proxyMsg, (time, Some(time)))))
        cache = cache ::: List(SendCache(proxyMsg, time, Some(time), Some(tunnel.toString)))
      }

  }

  def sendProxyMsg(msg: Msg, now: Long) =
    tunnel ! MsgWithTimeout(msg, now, msgTimeout)

  def startReConnTimer = context.system.scheduler.scheduleOnce(reconnInterval millis, self, Reconn)(context.system.dispatcher)
  def startCheckSendTimer = context.system.scheduler.scheduleOnce(checkMsgInterval millis, self, CheckSend)(context.system.dispatcher)
  override def preStart(): Unit = {
    createTunnel
    self ! CheckSend
  }

  def handleTimeout(now: Long, cache: SendCache): Option[SendCache]

  def handleCheckSend: Unit =
    {
      val now = System.currentTimeMillis()
      //println("before=" + cache.size)
      cache = cache.map {
        case pair @ SendCache(proxyMsg @ Msg(id, oldMsg, snd), initTime, lastSendTime, lastTarget) =>
          //case pair @ (id, (proxyMsg @ Msg(_, oldMsg, snd), (initTime, lastSendTime))) =>
          (if (lastSendTime == None || (now - lastSendTime.get) > msgTimeout) {
            handleTimeout(now, pair)
          } else
            Some(pair))
      }.filterNot(_ == None).map { _.get }
    //  println("after=" + cache.size)
    }

}

class ResendAckProxyAc(reconnInterval: Long, checkMsgInterval: Long, msgTimeout: Long, target: ActorRef) extends AckProxyAc(reconnInterval, checkMsgInterval, msgTimeout, target) {

  override def handleCheckSend = if (maybeOnline) super.handleCheckSend

  def handleTimeout(now: Long, cache: SendCache) = {
    sendProxyMsg(cache.msg, now)
    //Some((proxyMsg.id, (Msg(proxyMsg.id, proxyMsg.msg, proxyMsg.sender), (initTime, Some(now)))))
    Some(cache.copy(lastSendTime = Some(now), lastTarget = Some(tunnel.toString)))
  }
}

class RouteAckProxyAc(reconnInterval: Long, checkMsgInterval: Long, msgTimeout: Long, target: ActorRef) extends AckProxyAc(reconnInterval, checkMsgInterval, msgTimeout, target) {
  //println(self)
  def handleTimeout(now: Long, cache: SendCache) = {
    //println("context.parent=" + context.parent)
    println(s"$self timeMsg=${cache.msg} now=$now lastTime=${cache.lastSendTime} lastTarget=${cache.lastTarget}")
    context.parent.tell(cache.msg.msg, cache.msg.sender)
    None
  }
}
